package com.aotain.nyx.vc;

import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map.Entry;

import org.apache.commons.codec.binary.Base64;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;
import org.apache.log4j.Logger;

import com.aotain.common.mongo.DataColumn;
import com.aotain.apollo.mongodb.MongoImportTool;
import com.aotain.nyx.flink.BasicTuple;
import com.aotain.nyx.statis.AbnStatisTuple;

public class VCDetect implements FlatMapFunction<Tuple2<String,BasicTuple>, AbnStatisTuple>{

	/**
	  *
	  */
	private static final long serialVersionUID = -4708060362265904748L;
	
	/**
	 * 每分钟累积MAP
	 */
	private HashMap<String,Long> mapPostNum = new HashMap<String,Long>();
	

	/**
	 * 存放记录连续状态 最后的更新时间
	 */
	private HashMap<String,Long> mapLastTime = new HashMap<String,Long>();
	
	/**
	 * 记录攻击开始时间
	 */
	private HashMap<String,Long> mapAttackStart = new HashMap<String,Long>();
	
	//-----累积包，次数
	/**
	 * 持久-上行包字节数
	 */
	private HashMap<String,Long> mapStreamoctetsMem = new HashMap<String,Long>();
	
	/**
	 * 持久-上行包数
	 */
	private HashMap<String,Long> mapStreampacketMem = new HashMap<String,Long>();

	/**
	 * 持久-攻击次数
	 */
	private HashMap<String,Long> mapPostNumMem = new HashMap<String,Long>();
	
	/**
	 * 持续攻击时记录第一次开始的Map拼接的EntryKey
	 */
	private HashMap<String,String> mapEntryKey = new HashMap<String,String>();
	
	/**
	 * 攻击峰值
	 */
	private HashMap<String,Double> mapHighFlow = new HashMap<String,Double>();
	
	/**
	 * 结束等待时长
	 */
	private final long END_DURATION = 180*1000l;
	
	/**
	 * 连接次数门限 300次/秒
	 */
	private final int ATTACK_THRESHOLD = 5*60;

	@Override
	public void flatMap(Tuple2<String, BasicTuple> value,
			Collector<AbnStatisTuple> out) throws Exception {
		
//		String entryKey = entry.getKey();
		//mapWUrl.put(minKey, entry.getValue());
//		String[] tuplesp = entryKey.split("\\|",-1);
		BasicTuple tuple = value.f1;
		String dip = tuple.getDestIP();
		String sip = tuple.getSourceIP();
		String hashKey = dip + "_" + sip;
		String base64Key = hashKey;
		
		if(tuple.getSessionNum()>= ATTACK_THRESHOLD)
		{	
			// 
			//判断当前源->目的IP是否存在记录
			if(mapLastTime.containsKey(base64Key))
			{
				//存在记录，更新数据
				mapLastTime.put(base64Key, System.currentTimeMillis());
				Logger.getLogger(VCDetect.class).info("###VC ATTACK UPDATE" + hashKey);
				AbnStatisTuple reTuple = UpdateAttackLog(tuple);
				out.collect(reTuple);
			}
			else
			{
				//新发生的攻击行为，记录开始时间并且上报异常攻击日志，但是不上报统计数据
				
				mapLastTime.put(base64Key, System.currentTimeMillis());
				AbnStatisTuple reTuple = InsertNewAttackLog(tuple);
				out.collect(reTuple);
				Logger.getLogger(VCDetect.class).info("###VC ATTACK START" + hashKey);
			}
		}
		else
		{
			if(mapLastTime.containsKey(base64Key))
			{
				//如果存在记录，判断更新时间与当前时间差距是否超过了等待门限
				Long lastTime = mapLastTime.get(base64Key);
				if(System.currentTimeMillis() - lastTime > END_DURATION)
				{
					//超过30秒，认为攻击结束，记录结束时间，
					Logger.getLogger(VCDetect.class).info("###VC ATTACK END" + base64Key);
					EndAttackLog(hashKey,tuple.getUpStreamOctets(),tuple.getUpStreamPacket());
					//将数据从map中移除
					mapLastTime.remove(base64Key);
					mapStreamoctetsMem.remove(base64Key);
					mapStreampacketMem.remove(base64Key);
					mapPostNumMem.remove(base64Key);
					mapAttackStart.remove(base64Key);
					mapEntryKey.remove(base64Key);
					mapHighFlow.remove(base64Key);
				}
				else
				{
					//虽然未超过门限，但是在攻击等待范围时间内，将这一分钟的数据累加到攻击数据中。
					AbnStatisTuple reTuple = UpdateAttackLog(tuple);
					out.collect(reTuple);
				}
			}
		}
		
		
		// 对内存中存在的正在监控的攻击数据做核查判断是否定义攻击结束
			
		ArrayList<String> removeItem = new ArrayList<String>();
		for(Entry<String, Long> entry : mapLastTime.entrySet())
		{
			Long lastTime = entry.getValue();
			String entryLastTimeKey = entry.getKey();
			Logger.getLogger(VCDetect.class).info(
							String.format("###VC REMOVE hashkey %s time %d",base64Key,(System.currentTimeMillis() - lastTime)/1000));
					
			if(System.currentTimeMillis() - lastTime > END_DURATION)
			{
				//超过等待门限秒，认为攻击结束，记录结束时间，
				String entryKey = mapEntryKey.get(entryLastTimeKey);
				Logger.getLogger(VCDetect.class).info("###VC ATTACK END" + entryKey);
				EndAttackLog(entryKey,0l,0l);
				//将数据从map中移除
				removeItem.add(base64Key);
				mapStreamoctetsMem.remove(base64Key);
				mapStreampacketMem.remove(base64Key);
				mapHighFlow.remove(base64Key);
				mapAttackStart.remove(base64Key);
				mapEntryKey.remove(base64Key);	
				}
			}
				
			//remove 已经攻击结束的记录
			for(String rmoveKey : removeItem)
			{
				mapLastTime.remove(rmoveKey);
			}
				
			removeItem.clear();
			mapPostNum.clear();
	}
	
	
	private AbnStatisTuple InsertNewAttackLog(BasicTuple tuple)
	{
		String dip = tuple.getDestIP();
		String sip = tuple.getSourceIP();
		String idc = tuple.getIDC();
		
		String gis = new String(Base64.decodeBase64(tuple.getGIS()));
		String[] gisArray = gis.split("#",-1);
		
		String destAreaName = gisArray[0];
		String destGis = gisArray[1];
		String sourceAreaName = gisArray[2];
		String sourceGis = gisArray[3];
		String sourceAreaCountry = gisArray[4];
		String sourceAreaId = gisArray[5];							
		String sourceProvinceName =  gisArray[6].isEmpty()?sourceAreaCountry:gisArray[6]; //如果省为空，精确到国家
		String sourceAreaCityId = gisArray[7];
		String sourceAreaProvinceId = gisArray[8];
		
		String hashKey = dip + "_" + sip;

		String base64Key = hashKey; 
				
		long starttime = System.currentTimeMillis();
		mapAttackStart.put(base64Key,starttime);//记录开始时间
		
		String sAbnormalTable = "SDS_ABNORMAL_LOG2";
		String accesstime = new SimpleDateFormat("yyyyMMddHHmmss").format(new Date(starttime));
		String rowKeyAbnormal = dip + "_" + accesstime + "_" + sip + "_" + "13";	

		String desc = String.format("疑似暴力破解行为");
				
		MongoImportTool importtool = MongoImportTool.getInstance();
		List<DataColumn> row = new ArrayList<DataColumn>();
		row.add(new DataColumn("ROWKEY", rowKeyAbnormal)); //主键字段，用于更新记录
		row.add(new DataColumn("SOURCEIP", sip));
		row.add(new DataColumn("DESTPORT", ""));
		row.add(new DataColumn("ACCESSTIME", Long.parseLong(accesstime)));
		row.add(new DataColumn("ABRNORMAL", 13));
		row.add(new DataColumn("DESTIP", dip));
		row.add(new DataColumn("SOURCEAREA", sourceAreaName));
		row.add(new DataColumn("SOURCEGEO", sourceGis));
		row.add(new DataColumn("SOURCECOUNTRY", sourceAreaCountry));
		row.add(new DataColumn("DESTAREA", destAreaName));
		row.add(new DataColumn("DESTGEO", destGis));
		row.add(new DataColumn("ATTNUM", tuple.getUpStreamPacket()));
		row.add(new DataColumn("DESC", desc));
		row.add(new DataColumn("EVALUATE", "40"));
		row.add(new DataColumn("ENDTIME", accesstime));
		if(sourceProvinceName.equals("")) {
			sourceProvinceName = sourceAreaCountry;
		}
		row.add(new DataColumn("PROVINCE", sourceProvinceName));
		row.add(new DataColumn("UPSTREAMOCTETS", tuple.getUpStreamOctets()));
		row.add(new DataColumn("UPSTREAMPACKET", tuple.getUpStreamPacket()));
		row.add(new DataColumn("ATTTYPE", "2"));
		row.add(new DataColumn("FLOWDIRECTION", tuple.getFlowDirection()));
		
		importtool.InsertRowData(sAbnormalTable, row);
		
		mapStreamoctetsMem.put(base64Key, tuple.getUpStreamOctets());
		mapStreampacketMem.put(base64Key, tuple.getUpStreamPacket());
		
		
		AbnStatisTuple abnTuple = new AbnStatisTuple();
		abnTuple.setSourceIP(sip);
		abnTuple.setDestIP(dip);
		abnTuple.setType("VC");
		abnTuple.setUpStreamOctets(tuple.getUpStreamPacket());
		abnTuple.setUpStreamPacket(tuple.getUpStreamPacket());
		abnTuple.setAttackNum(1l);
		abnTuple.setSourceCountry(sourceAreaCountry);
		abnTuple.setSourceProvince(sourceProvinceName);
		abnTuple.setAttackType("2");
		abnTuple.setAttTime(accesstime);
		//		out.collect(abnTuple);
		
		long countValue = tuple.getUpStreamPacket();
						
		mapPostNumMem.put(base64Key, countValue);
		
		mapEntryKey.put(base64Key, base64Key);
		
		Logger.getLogger(VCDetect.class).info(
				String.format("###VC START ROWKEY:%s DESC:%s  HASHKEY:%s",rowKeyAbnormal, desc, hashKey));
		
		return abnTuple;
	}

	private AbnStatisTuple UpdateAttackLog(BasicTuple tuple)
	{
		String dip = tuple.getDestIP();
		String sip = tuple.getSourceIP();

		String hashKey = dip + "_" + sip;
		String base64Key = hashKey;
		String gis = new String(Base64.decodeBase64(tuple.getGIS()));
		String[] gisArray = gis.split("#",-1);
		
		String sourceAreaCountry = gisArray[4];					
		String sourceProvinceName =  gisArray[6].isEmpty()?sourceAreaCountry:gisArray[6]; 
		//String base64Key = new String(Base64.encodeBase64(hashKey.getBytes()));
		
		//url = new String(Base64.decodeBase64(url));	
		
		//开始时间
		//String accesstime = new SimpleDateFormat("yyyyMMddHHmmss").format(new Date(_mapAttackStart.get(base64Key)));
		//String rowKeyAbnormal = dip + "_" + accesstime + "_" + sip + "_" + "_3";
		
		//UDII数据特有
		long nUpStreamOctets = 0;
		long nUpstreampacket = 0;

		nUpStreamOctets = tuple.getUpStreamOctets();
		nUpstreampacket = tuple.getUpStreamPacket();
			
		if(mapStreamoctetsMem.containsKey(base64Key))
			mapStreamoctetsMem.put(base64Key, mapStreamoctetsMem.get(base64Key) + nUpStreamOctets);
		if(mapStreampacketMem.containsKey(base64Key))
			mapStreampacketMem.put(base64Key, mapStreampacketMem.get(base64Key) + nUpstreampacket);

		
		long countValue = tuple.getSessionNum();
//		if(_mapPostNumMem.containsKey(base64Key))
//			_mapPostNumMem.put(base64Key, _mapPostNumMem.get(base64Key) + countValue);
		
		String desc = String.format("疑似暴力破解行为");
		double maxValue = (double)nUpStreamOctets/(double)60/(double)1024 ;
		
		
		if(mapHighFlow.containsKey(base64Key))
		{
			double maxValueMem = mapHighFlow.get(base64Key);
			if(maxValueMem < maxValue)
			{
				maxValueMem = maxValue;
				mapHighFlow.put(base64Key, maxValueMem);
			}
			else
			{
				maxValue = maxValueMem;
			}
		}
		else
		{
			mapHighFlow.put(base64Key, maxValue);
		}
		
		desc = String.format("%s 攻击峰值为 :%f KB/s ", desc,maxValue);
		
		//开始时间
		String accesstime = new SimpleDateFormat("yyyyMMddHHmmss").format(new Date(mapAttackStart.get(base64Key)));
				
		String rowKeyAbnormal = dip + "_" + accesstime + "_" + sip + "_" + "13";
			
		MongoImportTool importtool = MongoImportTool.getInstance();
		String tableName = "SDS_ABNORMAL_LOG2";
		List<DataColumn> row = new ArrayList<DataColumn>();
		

		nUpStreamOctets = mapStreamoctetsMem.get(base64Key);
		nUpstreampacket = mapStreamoctetsMem.get(base64Key);
		row.add(new DataColumn("UPSTREAMOCTETS",nUpStreamOctets));
		row.add(new DataColumn("UPSTREAMPACKET",nUpstreampacket));
		desc = String.format("%s 攻击流量[%f KB]",desc, (double)nUpStreamOctets/(double)1024);
		
		
		
		desc = String.format("%s 攻击次数 [%d]",desc, countValue);
		Logger.getLogger(VCDetect.class).info(
				String.format("###VC ATTACK UPDATE ROWKEY:%s DESC:%s  HASHKEY:%s",rowKeyAbnormal, desc, hashKey));

		row.add(new DataColumn("DESC",desc));
		row.add(new DataColumn("ATTNUM",countValue));
		
		List<DataColumn> filter = new ArrayList<DataColumn>();
		filter.add(new DataColumn("ROWKEY",rowKeyAbnormal));
		importtool.UpdateRowData(tableName, filter, row);
		
		
		AbnStatisTuple abnTuple = new AbnStatisTuple();
		abnTuple.setDestIP(dip);
		abnTuple.setSourceIP(sip);
		abnTuple.setType("VC");
		nUpStreamOctets = tuple.getUpStreamOctets();
		nUpstreampacket = tuple.getUpStreamPacket();
		abnTuple.setUpStreamOctets(nUpStreamOctets);
		abnTuple.setUpStreamPacket(nUpstreampacket);
		abnTuple.setSourceCountry(sourceAreaCountry);
		abnTuple.setSourceProvince(sourceProvinceName);
		abnTuple.setAttackType("2");
		abnTuple.setAttTime(accesstime);
		return abnTuple;
		
		
//		collector.emit(new Values(dip,sip,"","",dateStr_d, dateStr_h, sourceAreaId, sourceAreaName,
//				sourceProvinceName, countValue, dateStr_hour,"VC", idc, sourceAreaCountry,(long)nUpStreamOctets,(long)nUpstreampacket));
		
		}
	
	private AbnStatisTuple EndAttackLog(String entryKey,long nUpStreamOctets,long nUpstreampacket)
	{
		//UpdateAttackLog(entry);


		String[] tuplesp = entryKey.split("_",-1);
		String dip = tuplesp[0];
		String sip = tuplesp[1];
		
		String hashKey = dip + "_" + sip;
		String base64Key = hashKey;

		//开始时间
		String accesstime = new SimpleDateFormat("yyyyMMddHHmmss").format(new Date(mapAttackStart.get(base64Key)));
		String endtime = new SimpleDateFormat("yyyyMMddHHmmss").format(new Date(mapLastTime.get(base64Key)+60*1000l));
		
		String rowKeyAbnormal = dip + "_" + accesstime + "_" + sip + "_" + "13";

		MongoImportTool importtool = MongoImportTool.getInstance();
		String tableName = "SDS_ABNORMAL_LOG2";
		List<DataColumn> row = new ArrayList<DataColumn>();
		
		
		String desc = String.format("疑似暴力破解行为");
		
		row.add(new DataColumn("ATTNUM", String.valueOf(mapPostNumMem.get(base64Key))));
		row.add(new DataColumn("ENDTIME", endtime));
		
		
		nUpStreamOctets = mapStreamoctetsMem.get(base64Key);
		nUpstreampacket = mapStreamoctetsMem.get(base64Key);
		row.add(new DataColumn("UPSTREAMOCTETS",nUpStreamOctets));
		row.add(new DataColumn("UPSTREAMPACKET",nUpstreampacket));
		desc = String.format("%s 攻击流量[%f KB]",desc, (double)nUpStreamOctets/(double)1024);
		
		
		long countValue = mapPostNumMem.get(base64Key);
		desc = String.format("%s 攻击次数 [%d]",desc, countValue);
		
		if(mapHighFlow.containsKey(base64Key))
		{
			double maxValueMem = mapHighFlow.get(base64Key);
			desc = String.format("%s 攻击峰值 [%f KB/s]",desc, maxValueMem);
		}
		
		row.add(new DataColumn("DESC",desc));
		row.add(new DataColumn("ATTNUM",countValue));
		
		List<DataColumn> filter = new ArrayList<DataColumn>();
		filter.add(new DataColumn("ROWKEY",rowKeyAbnormal));
		importtool.UpdateRowData(tableName, filter, row);
		
		AbnStatisTuple abnTuple = new AbnStatisTuple();
		abnTuple.setDestIP(dip);
		abnTuple.setSourceIP(sip);
		abnTuple.setType("VC");
		abnTuple.setUpStreamOctets(nUpStreamOctets);
		abnTuple.setUpStreamPacket(nUpstreampacket);
		abnTuple.setAttackType("2");
		abnTuple.setAttTime(accesstime);
		
		Logger.getLogger(VCDetect.class).info(
				String.format("###VC ATTACK END ROWKEY:%s DESC:%s  HASHKEY:%s",rowKeyAbnormal, desc, hashKey));
		return abnTuple;
	}
}





    